//! # 消息发布者
//!
//! 定义和实现消息发布者相关功能

use async_trait::async_trait;
use std::sync::Arc;

use crate::bus::{MessageBus, Publisher};
use crate::error::{MessageBusError, MessageBusResult};
use crate::message::Message;

/// 简单发布者实现
pub struct SimplePublisher {
    message_bus: Arc<dyn MessageBus>,
    publisher_id: String,
}

impl SimplePublisher {
    /// 创建新的发布者
    pub fn new(message_bus: Arc<dyn MessageBus>, publisher_id: String) -> Self {
        Self {
            message_bus,
            publisher_id,
        }
    }

    /// 获取发布者 ID
    pub fn id(&self) -> &str {
        &self.publisher_id
    }
}

#[async_trait]
impl Publisher for SimplePublisher {
    async fn publish(&self, message: Message) -> MessageBusResult<()> {
        self.message_bus.publish(message).await
    }

    async fn publish_batch(&self, messages: Vec<Message>) -> MessageBusResult<()> {
        for message in messages {
            self.publish(message).await?;
        }
        Ok(())
    }

    async fn publish_and_confirm(&self, message: Message) -> MessageBusResult<()> {
        // 对于简单实现，发布和确认是同一个操作
        self.publish(message).await
    }

    async fn close(&self) -> MessageBusResult<()> {
        // 简单实现中，发布者无需特殊关闭操作
        Ok(())
    }
}

/// 批处理发布者 - 支持批量发布优化
pub struct BatchPublisher {
    message_bus: Arc<dyn MessageBus>,
    publisher_id: String,
    batch_size: usize,
    batch_timeout_ms: u64,
    pending_messages: Arc<tokio::sync::Mutex<Vec<Message>>>,
    batch_sender: tokio::sync::mpsc::UnboundedSender<Message>,
}

impl BatchPublisher {
    /// 创建批处理发布者
    pub fn new(
        message_bus: Arc<dyn MessageBus>,
        publisher_id: String,
        batch_size: usize,
        batch_timeout_ms: u64,
    ) -> Self {
        let (batch_sender, mut batch_receiver) = tokio::sync::mpsc::unbounded_channel();
        let pending_messages = Arc::new(tokio::sync::Mutex::new(Vec::new()));
        
        // 启动批处理任务
        let bus_clone = message_bus.clone();
        let pending_clone = pending_messages.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(batch_timeout_ms));
            
            loop {
                tokio::select! {
                    // 接收新消息
                    Some(message) = batch_receiver.recv() => {
                        let mut pending = pending_clone.lock().await;
                        pending.push(message);
                        
                        // 如果达到批处理大小，立即发送
                        if pending.len() >= batch_size {
                            let messages = std::mem::take(&mut *pending);
                            drop(pending);
                            
                            if let Err(e) = Self::send_batch(&bus_clone, messages).await {
                                tracing::error!("批量发送失败: {}", e);
                            }
                        }
                    }
                    
                    // 定时发送
                    _ = interval.tick() => {
                        let mut pending = pending_clone.lock().await;
                        if !pending.is_empty() {
                            let messages = std::mem::take(&mut *pending);
                            drop(pending);
                            
                            if let Err(e) = Self::send_batch(&bus_clone, messages).await {
                                tracing::error!("定时批量发送失败: {}", e);
                            }
                        }
                    }
                }
            }
        });

        Self {
            message_bus,
            publisher_id,
            batch_size,
            batch_timeout_ms,
            pending_messages,
            batch_sender,
        }
    }

    async fn send_batch(bus: &Arc<dyn MessageBus>, messages: Vec<Message>) -> MessageBusResult<()> {
        let batch_size = messages.len();
        let start_time = std::time::Instant::now();
        
        tracing::debug!("开始批量发送消息: 数量={}", batch_size);
        
        let mut success_count = 0;
        let mut failed_count = 0;
        
        for (index, message) in messages.into_iter().enumerate() {
            let topic = message.topic().to_string();
            match bus.publish(message).await {
                Ok(()) => {
                    success_count += 1;
                    tracing::trace!("批量消息 #{} 发送成功: topic={}", index + 1, topic);
                }
                Err(e) => {
                    failed_count += 1;
                    tracing::warn!("批量消息 #{} 发送失败: topic={}, error={}", index + 1, topic, e);
                }
            }
        }
        
        let duration = start_time.elapsed();
        
        if failed_count > 0 {
            tracing::warn!(
                "批量发送部分失败: 成功={}, 失败={}, duration={:?}",
                success_count,
                failed_count,
                duration
            );
            Err(MessageBusError::publish_error(
                "batch",
                &format!("批量发送部分失败: {} 条消息失败", failed_count)
            ))
        } else {
            tracing::info!(
                "批量发送成功: 数量={}, duration={:?}",
                success_count,
                duration
            );
            Ok(())
        }
    }

    /// 获取发布者 ID
    pub fn id(&self) -> &str {
        &self.publisher_id
    }

    /// 获取批处理大小
    pub fn batch_size(&self) -> usize {
        self.batch_size
    }

    /// 获取批处理超时时间
    pub fn batch_timeout_ms(&self) -> u64 {
        self.batch_timeout_ms
    }

    /// 强制发送当前批次
    pub async fn flush(&self) -> MessageBusResult<()> {
        let mut pending = self.pending_messages.lock().await;
        if !pending.is_empty() {
            let messages = std::mem::take(&mut *pending);
            drop(pending);
            Self::send_batch(&self.message_bus, messages).await?;
        }
        Ok(())
    }
}

#[async_trait]
impl Publisher for BatchPublisher {
    async fn publish(&self, message: Message) -> MessageBusResult<()> {
        self.batch_sender.send(message)
            .map_err(|_| MessageBusError::publish_error("batch", "发送到批处理队列失败"))?;
        Ok(())
    }

    async fn publish_batch(&self, messages: Vec<Message>) -> MessageBusResult<()> {
        for message in messages {
            self.publish(message).await?;
        }
        Ok(())
    }

    async fn publish_and_confirm(&self, message: Message) -> MessageBusResult<()> {
        // 对于批处理发布者，需要等待实际发送完成
        self.message_bus.publish(message).await
    }

    async fn close(&self) -> MessageBusResult<()> {
        // 发送剩余的消息
        self.flush().await?;
        Ok(())
    }
}

/// 发布者构建器
pub struct PublisherBuilder {
    publisher_id: Option<String>,
    batch_size: Option<usize>,
    batch_timeout_ms: Option<u64>,
    enable_batching: bool,
}

impl Default for PublisherBuilder {
    fn default() -> Self {
        Self::new()
    }
}

impl PublisherBuilder {
    /// 创建新的发布者构建器
    pub fn new() -> Self {
        Self {
            publisher_id: None,
            batch_size: None,
            batch_timeout_ms: None,
            enable_batching: false,
        }
    }

    /// 设置发布者 ID
    pub fn id(mut self, id: impl Into<String>) -> Self {
        self.publisher_id = Some(id.into());
        self
    }

    /// 启用批处理
    pub fn enable_batching(mut self) -> Self {
        self.enable_batching = true;
        self
    }

    /// 设置批处理大小
    pub fn batch_size(mut self, size: usize) -> Self {
        self.batch_size = Some(size);
        self.enable_batching = true;
        self
    }

    /// 设置批处理超时时间
    pub fn batch_timeout_ms(mut self, timeout_ms: u64) -> Self {
        self.batch_timeout_ms = Some(timeout_ms);
        self.enable_batching = true;
        self
    }

    /// 构建发布者
    pub fn build(self, message_bus: Arc<dyn MessageBus>) -> Box<dyn Publisher> {
        let publisher_id = self.publisher_id.unwrap_or_else(|| {
            format!("publisher-{}", uuid::Uuid::new_v4())
        });

        if self.enable_batching {
            let batch_size = self.batch_size.unwrap_or(100);
            let batch_timeout_ms = self.batch_timeout_ms.unwrap_or(1000);
            
            Box::new(BatchPublisher::new(
                message_bus,
                publisher_id,
                batch_size,
                batch_timeout_ms,
            ))
        } else {
            Box::new(SimplePublisher::new(message_bus, publisher_id))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::memory::InMemoryMessageBus;
    use serde_json::json;

    #[tokio::test]
    async fn test_simple_publisher() {
        let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
        let publisher = SimplePublisher::new(bus.clone(), "test-publisher".to_string());

        assert_eq!(publisher.id(), "test-publisher");

        let message = Message::new("test.topic", json!({"test": "data"}));
        assert!(publisher.publish(message).await.is_ok());
    }

    #[tokio::test]
    async fn test_batch_publisher() {
        let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
        let publisher = BatchPublisher::new(
            bus.clone(),
            "batch-publisher".to_string(),
            2,  // 批处理大小
            100, // 100ms 超时
        );

        assert_eq!(publisher.id(), "batch-publisher");
        assert_eq!(publisher.batch_size(), 2);
        assert_eq!(publisher.batch_timeout_ms(), 100);

        // 发送消息
        let message1 = Message::new("test.topic", json!({"id": 1}));
        let message2 = Message::new("test.topic", json!({"id": 2}));
        
        assert!(publisher.publish(message1).await.is_ok());
        assert!(publisher.publish(message2).await.is_ok());

        // 等待批处理完成
        tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;

        // 强制刷新
        assert!(publisher.flush().await.is_ok());
    }

    #[tokio::test]
    async fn test_publisher_builder() {
        let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
        
        // 简单发布者
        let simple_publisher = PublisherBuilder::new()
            .id("simple-test")
            .build(bus.clone());
        
        let message = Message::new("test.topic", json!({"test": "data"}));
        assert!(simple_publisher.publish(message).await.is_ok());

        // 批处理发布者
        let batch_publisher = PublisherBuilder::new()
            .id("batch-test")
            .enable_batching()
            .batch_size(10)
            .batch_timeout_ms(500)
            .build(bus.clone());

        let message = Message::new("test.topic", json!({"test": "batch"}));
        assert!(batch_publisher.publish(message).await.is_ok());
    }
}